package com.lagou

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object WorkThree {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getCanonicalName}")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext

    // 读取click文件生成RDD
    val clickRDD: RDD[(String, (Int, Int))] = sc.textFile("data/source3/click.log").map {
      line =>
        val arr = line.split("\\s+")
        val adId = arr(3).substring(arr(3).lastIndexOf("=") + 1)
        (adId, (1, 0))
    }

    // 读取imp文件生成RDD
    val impRDD: RDD[(String, (Int, Int))] = sc.textFile("data/source3/imp.log").map {
      line =>
        val arr = line.split("\\s+")
        val adId = arr(3).substring(arr(3).lastIndexOf("=") + 1)
        (adId, (0, 1))
    }

    // join
    val finalRDD: RDD[(String, (Int, Int))] = clickRDD.union(impRDD)
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))

    finalRDD.foreach(println)

    // 写hdfs
    finalRDD.saveAsTextFile("hdfs://linux121:9000/test/clickImpCount/")

    spark.close()
  }
}
